package com.garlicts.framework.registry;

import com.garlicts.framework.core.FrameworkConstant;
import com.garlicts.framework.annotation.Bean;
import com.garlicts.framework.config.PropertiesProvider;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * zk服务注册
 *
 */
@Bean
public class ZookeeperServiceRegistry implements ServiceRegistry {

	private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperServiceRegistry.class);

	private CuratorFramework zkClient;

	public ZookeeperServiceRegistry() {
		String zkAddressStr = PropertiesProvider.getString(FrameworkConstant.CUSTOM_ZOOKEEPER_ADDRESS);
		String[] zkAddressArr = zkAddressStr.split(",");
		String zkAddress = zkAddressArr[0];
		zkClient = CuratorFrameworkFactory.builder().connectString(zkAddress)
				.sessionTimeoutMs(Constant.ZK_SESSION_TIMEOUT).connectionTimeoutMs(Constant.ZK_CONNECTION_TIMEOUT)
				.retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
		zkClient.start();
		LOGGER.debug("connect zookeeper");
	}

	@Override
	public void register(String serviceAddress) throws Exception {
		// 创建 registry 节点（持久）
		String registryPath = Constant.ZK_REGISTRY_PATH;
		Stat stat = zkClient.checkExists().forPath(registryPath);
		if (stat == null) {
			zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(registryPath);
		}

		// 创建 address 节点
		String nodePath = registryPath + "/" + serviceAddress;
		Stat nodePathStat = zkClient.checkExists().forPath(nodePath);
		if(nodePathStat != null){
			zkClient.delete().forPath(nodePath);
		}

		String addressNode = zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(nodePath,
				serviceAddress.getBytes());
		LOGGER.debug("create rpc server address node: {}", addressNode);

//		byte[] bytes = zkClient.getData().forPath(addressNode);
//		System.out.println(new String(bytes));

	}

	public static void main(String[] args) throws Exception {
		new ZookeeperServiceRegistry().register("127.0.0.1:6060");
	}

}
